package com.htiiot.hbase;
import java.io.IOException;
import java.util.Arrays;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.junit.Test;

/**
 * 本类用于从tsdb-uid表中获取metric并从metric中解析出did
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class GetDid {
    private static final Log LOG = LogFactory.getLog(org.apache.hadoop.hbase.mapreduce.Export.class);
    final static String NAME = "test";
    final static String RAW_SCAN = "hbase.mapreduce.include.deleted.rows";
    final static String EXPORT_BATCHING = "hbase.export.scanner.batch";

    static class ExportTableMapper
            extends TableMapper<ImmutableBytesWritable, Result> {

        @Override
        public void map(ImmutableBytesWritable key, Result value, Context context)
                throws IOException, InterruptedException {

        }

    }

    public static Job createSubmittableJob(Configuration conf, String[] args)
            throws IOException {
        String tableName = "tsdb-uid";
        Path outputDir = new Path(args[0]);
        Job job = new Job(conf, NAME);
        job.setJarByClass(GetDid.class);
        // Set optional scan parameters
        Scan s = getConfiguredScanForJob(conf, args);
        IdentityTableMapper.initJob(tableName, s, ExportTableMapper.class, job);
        // No reducers.  Just write straight to output files.
        job.setNumReduceTasks(0);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(Result.class);
        FileOutputFormat.setOutputPath(job, outputDir); // job conf doesn't contain the conf so doesn't have a default fs.
        return job;
    }

    private static Scan getConfiguredScanForJob(Configuration conf, String[] args) throws IOException {
        Scan s = new Scan();
        // Optional arguments.
        // Set Scan Versions
        int versions = args.length > 2? Integer.parseInt(args[2]): 1;
        s.setMaxVersions(versions);
        // Set Scan Range
        long startTime = args.length > 3? Long.parseLong(args[3]): 0L;
        long endTime = args.length > 4? Long.parseLong(args[4]): Long.MAX_VALUE;
        s.setTimeRange(startTime, endTime);
        // Set cache blocks
        s.setCacheBlocks(false);
        // set Start and Stop row
        if (conf.get(TableInputFormat.SCAN_ROW_START) != null) {
            s.setStartRow(Bytes.toBytes(conf.get(TableInputFormat.SCAN_ROW_START)));
        }
        if (conf.get(TableInputFormat.SCAN_ROW_STOP) != null) {
            s.setStopRow(Bytes.toBytes(conf.get(TableInputFormat.SCAN_ROW_STOP)));
        }
        // Set Scan Column Family
        boolean raw = Boolean.parseBoolean(conf.get(RAW_SCAN));
        if (raw) {
            s.setRaw(raw);
        }

        if (conf.get(TableInputFormat.SCAN_COLUMN_FAMILY) != null) {
            s.addFamily(Bytes.toBytes(conf.get(TableInputFormat.SCAN_COLUMN_FAMILY)));
        }
        // Set RowFilter or Prefix Filter if applicable.
        Filter exportFilter = getExportFilter(args);
        if (exportFilter!= null) {
            LOG.info("Setting Scan Filter for Export.");
            s.setFilter(exportFilter);
        }

        int batching = conf.getInt(EXPORT_BATCHING, -1);
        if (batching !=  -1){
            try {
                s.setBatch(batching);
            } catch (IncompatibleFilterException e) {
                LOG.error("Batching could not be set", e);
            }
        }
        LOG.info("versions=" + versions + ", starttime=" + startTime +
                ", endtime=" + endTime + ", keepDeletedCells=" + raw);
        return s;
    }

    private static Filter getExportFilter(String[] args) {
        Filter exportFilter = null;
        String filterCriteria = (args.length > 5) ? args[5]: null;
        if (filterCriteria == null) return null;
        if (filterCriteria.startsWith("^")) {
            String regexPattern = filterCriteria.substring(1, filterCriteria.length());
            exportFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(regexPattern));
        } else {
            exportFilter = new PrefixFilter(Bytes.toBytes(filterCriteria));
        }
        return exportFilter;
    }

    /*
     * @param errorMsg Error message.  Can be null.
     */
    private static void usage(final String errorMsg) {
        if (errorMsg != null && errorMsg.length() > 0) {
            System.err.println("ERROR: " + errorMsg);
        }
        System.err.println("Usage: Export [-D <property=value>]* <tablename> <outputdir> [<versions> " +
                "[<starttime> [<endtime>]] [^[regex pattern] or [Prefix] to filter]]\n");
        System.err.println("  Note: -D properties will be applied to the conf used. ");
        System.err.println("  For example: ");
        System.err.println("   -D mapreduce.output.fileoutputformat.compress=true");
        System.err.println("   -D mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.GzipCodec");
        System.err.println("   -D mapreduce.output.fileoutputformat.compress.type=BLOCK");
        System.err.println("  Additionally, the following SCAN properties can be specified");
        System.err.println("  to control/limit what is exported..");
        System.err.println("   -D " + TableInputFormat.SCAN_COLUMN_FAMILY + "=<familyName>");
        System.err.println("   -D " + RAW_SCAN + "=true");
        System.err.println("   -D " + TableInputFormat.SCAN_ROW_START + "=<ROWSTART>");
        System.err.println("   -D " + TableInputFormat.SCAN_ROW_STOP + "=<ROWSTOP>");
        System.err.println("For performance consider the following properties:\n"
                + "   -Dhbase.client.scanner.caching=100\n"
                + "   -Dmapreduce.map.speculative=false\n"
                + "   -Dmapreduce.reduce.speculative=false");
        System.err.println("For tables with very wide rows consider setting the batch size as below:\n"
                + "   -D" + EXPORT_BATCHING + "=10");
    }

    //arg[0]指定输出路径
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            usage("Wrong number of arguments: " + otherArgs.length);
            System.exit(-1);
        }
        Job job = createSubmittableJob(conf, otherArgs);
        System.exit(job.waitForCompletion(true)? 0 : 1);
    }


    @Test
    public void getTid() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "192.168.0.85:2181");
//        conf.set("zookeeper.znode.parent","/hbase-unsecure");
        Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(TableName.valueOf("tsdb-cs"));
        Result r = table.get(new Get("0000003C00004229000000000001977B_pe".getBytes()));
//        Cell cell = r.getColumnLatestCell("id".getBytes(), "tagv".getBytes());
//        System.out.println(Arrays.toString(CellUtil.cloneValue(cell)));
        for(Cell value:r.listCells()){
                String rowKey = Bytes.toStringBinary(CellUtil.cloneRow(value));
                String familyName = Bytes.toStringBinary(CellUtil.cloneFamily(value));
                String qualifierName = Bytes.toStringBinary(CellUtil.cloneQualifier(value));
                String v = Bytes.toStringBinary(CellUtil.cloneValue(value));
                System.out.println(Arrays.toString(CellUtil.cloneRow(value)));
                System.out.println(rowKey+"|"+familyName+"|"+qualifierName+"|"+v);
            }
        conn.close();
    }

    @Test
    public void getRowKey() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "192.168.0.83:2181");
        conf.set("zookeeper.znode.parent","/hbase-unsecure");
        Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(TableName.valueOf("tsdb"));
        Scan s = new Scan().setFilter(new PageFilter(1));
        ResultScanner rs = table.getScanner(s);
        rs.forEach(x->{
            byte[] row = x.getRow();
            System.out.println(Arrays.toString(row));
            System.out.println(row.length);
        });
        conn.close();
    }

    @Test
    public void t() throws IOException {
        String[] args = new String[]{"-Dmapreduce.output.fileoutputformat.compress=true","b"};
        String[] r = new GenericOptionsParser(HBaseConfiguration.create(), args).getRemainingArgs();
        System.out.println(Arrays.toString(r));
    }

    @Test
    public void other(){
        byte[] a = {1, 2, 3};
        byte[] b = {1, 2, 3};
        System.out.println(Arrays.toString(Arrays.copyOfRange(a,0,2)));
    }

}